package com.example.wac.config;


import com.example.wac.listener.MqListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * @author wuancheng
 * @description
 * @date 2021-07-28 13:47:00
 */

@Slf4j
@Configuration
public class RocketMqConfig {
    /**
     * 消费者组
     */
    @Value("${rocketMq.CONSUMER_GROUP}")
    private String consumerGroup;

    /**
     * 生产者组
     */
    @Value("${rocketMq.PRODUCER_Group}")
    private String producerGroup;

    @Autowired
    MqListener mqListener;


    /**
     * Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端，该消费模式一般实时性较高
     *
     * @param nameServer 名字服务地址
     * @param topic      主题
     * @return DefaultMQPushConsumer
     * @throws MQClientException
     */
    @Bean
    public DefaultMQPushConsumer mqPushConsumer(@Value("${rocketMq.NAME_SERVER}") String nameServer
            , @Value("${rocketMq.TOPIC}") String topic
            , @Value("${rocketMq.TOPIC1}") String topic1) throws MQClientException {
        // 构造一个确定消费者组的消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        // 配置名字服务地址
        consumer.setNamesrvAddr(nameServer);
        // 消费者模式 默认集群模式CLUSTERING 同一个消费者集群中订阅同一个主题，一个消息只会被一个消费者消费
        // 广播模式BROADCASTING 消息会发送给消费者组的每一个消费者进行消费
        consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 消费模式:
         * CONSUME_FROM_LAST_OFFSET 一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
         * CONSUME_FROM_FIRST_OFFSET 从queue的第一条消息开始消费
         * CONSUME_FROM_TIMESTAMP 从具体的时间戳开始消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * push 推动型消费
         * 订阅主题和 标签（ * 代表所有标签)下信息 可以过滤消息
         * 注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
         */
        consumer.subscribe(topic, "mailTag");
        consumer.subscribe(topic1, "*");
        consumer.registerMessageListener(mqListener);
        // 自建服务 AccessChannel.CLOUD会在TraceConstants.TRACE_TOPIC_PREFIX加上regionId作为traceTopic
        consumer.setAccessChannel(AccessChannel.LOCAL);
        // 启动消费端
        consumer.start();
        if (log.isDebugEnabled()) {
            log.info("消费者 启动成功=======");
        }
        return consumer;
    }

//    @Bean(name = "mqPushConsumer1")
//    public DefaultMQPushConsumer mqPushConsumer1(@Value("${rocketMq.NAME_SERVER}") String nameServer
//            , @Value("${rocketMq.TOPIC}") String topic
//            , @Value("${rocketMq.TOPIC1}") String topic1) throws MQClientException {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//        consumer.setNamesrvAddr(nameServer);
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        consumer.start();
//        /**
//         * pull 拉取型消费
//         */
//        Set<MessageQueue> queueSet = consumer.fetchSubscribeMessageQueues("mailTag");
//        Map<MessageQueue, Long> offsetTable = new HashMap<MessageQueue, Long>();
//        for (MessageQueue mq : queueSet) {
//
//        }
//        return consumer;
//    }

    /**
     * 生产者
     * @param nameServer 名字服务地址
     * @return
     * @throws MQClientException
     */
    @Bean
    public DefaultMQProducer mqProducer(@Value("${rocketMq.NAME_SERVER}")String nameServer) throws MQClientException {
        //构造一个确定生产者组的生产者
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        //不开启vip通道 开通口端口会减2
        producer.setVipChannelEnabled(false);
        //配置名字服务地址
        producer.setNamesrvAddr(nameServer);
        //设置生产者实例名称
        producer.setInstanceName(producerGroup);
        //设置失败重试次数
        producer.setRetryTimesWhenSendFailed(5);
        //启动生产端
        producer.start();
        if (log.isDebugEnabled()){
            log.info("生产者 启动成功=======");
        }
        return producer;
    }

}
